package com.foreveross.fetl.web.action.preprocess;

import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Controller;

import com.foreveross.fetl.common.hbase.client.HBaseHelper;
import com.foreveross.fetl.core.domain.task.SourceEntity;
import com.foreveross.fetl.util.DateUtils;
import com.foreveross.fetl.util.RowkeyUtils;
import com.foreveross.fetl.web.action.BaseAction;

@Controller("chaneRowKeyAction")
@Scope("prototype")
public class ChaneRowKeyAction extends BaseAction {

	/**
	 * 
	 */
	private static final long serialVersionUID = -445818729757547181L;
	
	public String list() {
		return "METHOD";
	}
	
	public String manualExecute(){
		String tableName = getRequest().getParameter("tableName");
		SourceEntity source = new SourceEntity();
		List<SourceEntity> listHome=source.findSourceEntityByType(SourceEntity.TYPE_HOME);
		List<SourceEntity> listchannels = source.findSourceEntityByType(SourceEntity.TYPE_CHNNAEL);
		List<String> strchannels= new ArrayList<String>();
		if(null!=listHome){
			for(SourceEntity entity:listHome){
				strchannels.add(entity.getHbaseKey());
			}
		}
		if(null!=listchannels){
			for(SourceEntity entity:listchannels){
				strchannels.add(entity.getHbaseKey());
			}
		}
		if(tableName.startsWith("LP_FLIGHT_HISTORY")){
			for(String channel :strchannels){
				String startRowkey =this.getStartRowkey(null,channel,tableName);
				String endRowKey=this.getEndRowkey(null,channel,tableName);
				this.processByHour(startRowkey,endRowKey,tableName);
			}
		}else{
			String start=getRequest().getParameter("start");
			String end =getRequest().getParameter("end");
			Date startTime =DateUtils.parse(start+"00", "yyyyMMddHH");
			Date endTime =DateUtils.parse(end+"23", "yyyyMMddHH");
			Calendar cStart = Calendar.getInstance();
			cStart.setTime(startTime);
			Calendar cEnd = Calendar.getInstance();
			cEnd.setTime(endTime);
			cEnd.add(Calendar.HOUR_OF_DAY, +1);
			while (cStart.before(cEnd)) {
				String grabTimeStr=DateUtils.format(cStart.getTime(),"yyyyMMddHH");
				for(String channel :strchannels){
					String startRowkey =this.getStartRowkey(grabTimeStr,channel,tableName);
					String endRowKey=this.getEndRowkey(grabTimeStr,channel,tableName);
					this.processByHour(startRowkey,endRowKey,tableName);
				}
				cStart.add(Calendar.HOUR_OF_DAY, 1);// 小时数加一
			}
		}
		
		return "JSON";
	}
	
	private void processByHour(String startRowkey, String endRowKey,
			String tableName) {
		ResultScanner rs = null;
		try {
			rs = HBaseHelper.getRecords(tableName,startRowkey, endRowKey);
		} catch (Exception e1) {
			
		}
		Map<String, Map<String, String>> currentMap = new HashMap<String, Map<String, String>>();// 保存HBASE中存储的信息
		try{
			for (Result res : rs) {
				try{
					 currentMap = HBaseHelper.transformHBaseRowToMap(res);
					 String newRowkey=this.getNewRowkey(currentMap,tableName);
					 this.saveToChaneRowKey(tableName, newRowkey, currentMap);
				} catch (Exception e) {
					
				}
			}
		}finally{
			if(rs != null) rs.close();
			rs = null;
		}
	}

	private String getNewRowkey(Map<String, Map<String, String>> fcv,
			String tableName) {
		Map<String, String> currentMap = fcv.get("planeinfo");
		String grabDate=currentMap.get("grabDate").substring(0,10);
		String grabChannelNum = currentMap.get("grabChannelNum");
		String fromCity = currentMap.get("fromCity");
		String toCity = currentMap.get("toCity");
		if(currentMap.get("flightId")==null) return null;
		if(currentMap.get("flightId").indexOf("|")>0){
			currentMap.put("flightId", currentMap.get("flightId").substring(0, currentMap.get("flightId").indexOf("|")));
		}
		if(currentMap.get("flightId").indexOf(",")>0){
			currentMap.put("flightId", currentMap.get("flightId").substring(0, currentMap.get("flightId").indexOf(",")));
		}
		String flightId = RowkeyUtils.getFlightIdHbaseKey(currentMap
				.get("flightId"));
		String flightBatchStartTime = currentMap.get("flightBatchStartTime")
				.substring(0, 10);
		String rowKey=null;
		if(tableName.startsWith("CRAWL_PLANE_FREIGHT")){
			rowKey=RowkeyUtils.getRowKey(grabDate, grabChannelNum, fromCity, toCity, flightBatchStartTime, flightId, RowkeyUtils.START_ROW_KEY);
			rowKey=rowKey.substring(0, rowKey.length()-5)+this.random();
		}
		if(tableName.equals("LP_FLIGHT_HISTORY")){
			StringBuffer newRowkey =new StringBuffer();
			newRowkey.append(grabChannelNum);
			newRowkey.append(fromCity);
			newRowkey.append(toCity);
			newRowkey.append(grabDate.substring(0, 8));
			newRowkey.append(flightBatchStartTime);
			newRowkey.append(flightId);
			rowKey=newRowkey.toString();
		}
		if(tableName.equals("LP_FLIGHT_HISTORY_DOUBLE")){
			String flightReturnTime=null;
			boolean isReturn=false;
			for(String infos : fcv.keySet()){
				if(infos.equalsIgnoreCase("returnPlaneInfosInfo")){
					isReturn=true;
					break;
				}
			}
			if(!isReturn){
				return null;
			}
			Map<String,String> returnMap=fcv.get("returnPlaneInfosInfo");
			for(String key:returnMap.keySet()){
				if(key.startsWith("flightBatchStartTime")){
					flightReturnTime=returnMap.get(key).substring(0,8);
					break;
				}
				if(key.startsWith("flightBatchDate")){
					flightReturnTime=returnMap.get(key).replace("-", "").substring(0,8);
					break;
				}
			}
			StringBuffer transitFlightNo = new StringBuffer();
			boolean isReturnTransit=false;
			for(String infos : fcv.keySet()){
				if(infos.equalsIgnoreCase("transitsInfo")){
					isReturnTransit=true;
					break;
				}
			}
			if(isReturnTransit){
				Map<String, String> transit =fcv.get("transitsInfo");

				for(String key : transit.keySet()){
					if(key.startsWith("flightNo")){
						String flightNo=transit.get(key);
						if(flightNo.indexOf("|")>0){
							flightNo=flightNo.substring(0, flightNo.indexOf("|"));
						}
						if(flightNo.indexOf(",")>0){
							flightNo=flightNo.substring(0, flightNo.indexOf(","));
						}
						if(flightId.equals(flightNo))
							continue;
						transitFlightNo.append(RowkeyUtils.getFlightIdHbaseKey(flightNo));
					}
				}
			}
			if(grabChannelNum==null || fromCity==null || toCity==null || flightReturnTime==null || grabDate==null){
				return null;
			}
			for (int j = transitFlightNo.length(); j < 16; j++) {
				transitFlightNo.append("0");
			}
			StringBuffer newRowkey =new StringBuffer();
			newRowkey.append(grabChannelNum);
			newRowkey.append(fromCity);
			newRowkey.append(toCity);
			newRowkey.append(grabDate.substring(0, 8));
			newRowkey.append(flightBatchStartTime);
			newRowkey.append(flightReturnTime);
			newRowkey.append(flightId);
			newRowkey.append(transitFlightNo);
			rowKey=newRowkey.toString();
		}
		
		return rowKey;
	}
	

	private String random() {
		Integer n  = (int)(Math.random()*100000);
		return n.toString();
	}

	private String getEndRowkey(String grabTimeStr, String channel,
			String tableName) {
		StringBuffer startRowKey =new StringBuffer();
		String rowkey=null;
		if(tableName.startsWith("CRAWL_PLANE_FREIGHT")){
			startRowKey.append(grabTimeStr);
			startRowKey.append(channel);
			rowkey=RowkeyUtils.appendRowkey(startRowKey.toString(), 46, RowkeyUtils.STOP_ROW_KEY);
		}
		if(tableName.startsWith("LP_FLIGHT_HISTORY")){
			startRowKey.append(channel);
			startRowKey.append("99");
			rowkey=RowkeyUtils.appendRowkey(startRowKey.toString(), 37, RowkeyUtils.STOP_ROW_KEY);
		}
		return rowkey;
	}

	private String getStartRowkey(String grabTimeStr, String channel,
			String tableName) {
		StringBuffer startRowKey =new StringBuffer();
		String rowkey=null;
		if(tableName.startsWith("CRAWL_PLANE_FREIGHT")){
			startRowKey.append(grabTimeStr);
			startRowKey.append(channel);
			rowkey=RowkeyUtils.appendRowkey(startRowKey.toString(), 46, RowkeyUtils.START_ROW_KEY);
		}
		if(tableName.startsWith("LP_FLIGHT_HISTORY")){
			startRowKey.append(channel);
			rowkey=RowkeyUtils.appendRowkey(startRowKey.toString(), 37, RowkeyUtils.START_ROW_KEY);
		}
		return rowkey;
	}

	private void saveToChaneRowKey(String tableName,String rowKey,Map<String, Map<String, String>> fcv){
		for (Entry<String, Map<String, String>> fe : fcv.entrySet()) {
			Put put = new Put(rowKey.getBytes());
			for (Entry<String, String> ce : fe.getValue().entrySet()) {
				put.add(fe.getKey().getBytes(), ce.getKey().getBytes(), ce
						.getValue().getBytes());
			}
			try {
				HBaseHelper.addRecord(tableName, put);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		
	}
	

}
